package com.xiaoxx.rabbitmq.broker;


import com.alibaba.fastjson.JSONObject;
import com.xiaoxx.config.ThreadUtils;
import com.xiaoxx.rabbitmq.bean.CorrelationDataExt;
import com.xiaoxx.rabbitmq.bean.MyMessage;
import com.xiaoxx.rabbitmq.config.RabbitTemplateContainer;
import com.xiaoxx.rabbitmq.constant.BrokerMessageStatus;
import com.xiaoxx.rabbitmq.constant.MessageType;
import com.xiaoxx.rabbitmq.db.entity.BrokerMessage;
import com.xiaoxx.rabbitmq.db.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Objects;

/**
 * $RabbitBrokerImpl 真正的发送不同类型的消息实现类
 *
 * @author quhiu
 */
@Slf4j
@Component
public class RabbitBrokerImpl implements RabbitBroker {

    @Autowired
    private RabbitTemplateContainer rabbitTemplateContainer;

    @Autowired
    private MessageStoreService messageStoreService;

    /**
     * 迅速消息发送
     *
     * @param myMessage 消息
     */
    @Override
    public void rapidSend(MyMessage myMessage) {
        myMessage.setMessageType(MessageType.RAPID);
        sendKernel(myMessage);
    }


    /**
     * 确认消息发送
     *
     * @param myMessage 消息
     */
    @Override
    public void confirmSend(MyMessage myMessage) {
        myMessage.setMessageType(MessageType.CONFIRM);
        sendKernel(myMessage);
    }

    /**
     * 可靠新消息的发送
     *
     * @param myMessage 消息
     */
    @Override
    public void reliantSend(MyMessage myMessage) {
        try {
            myMessage.setMessageType(MessageType.RELIANT);
            BrokerMessage bm = messageStoreService.getById(myMessage.getMessageId());
            if (bm == null) {
                // 1. 把数据库的消息发送日志先记录好
                LocalDateTime now = LocalDateTime.now();
                BrokerMessage brokerMessage = new BrokerMessage();
                brokerMessage.setMessageId(myMessage.getMessageId());
                brokerMessage.setStatus(BrokerMessageStatus.SENDING.getCode());
                brokerMessage.setTryCount(0);
                brokerMessage.setCopCount(0);
                // tryCount 在最开始发送的时候不需要进行设置
                // brokerMessage.setNextRetry(now.plusMinutes(RabbitConstant.TIMEOUT));
                brokerMessage.setCreateTime(now);
                brokerMessage.setUpdateTime(now);
                brokerMessage.setMyMessage(myMessage);
                messageStoreService.save(brokerMessage);
            }
            // 2. 执行真正的发送消息逻辑
            sendKernel(myMessage);
        } catch (DataAccessException e) {
            log.error("#发送MQ新消息失败 数据库无法访问 可靠消息无法入库#, 消息:{} 异常:{}", JSONObject.toJSONString(myMessage), e.getMessage(), e);
        } catch (Exception e) {
            log.error("#发送MQ新消息失败#, 消息:{} 异常:{}", JSONObject.toJSON(myMessage), e.getMessage(), e);
            if (Objects.equals(myMessage.getMessageType(), MessageType.RELIANT)) {
                messageStoreService.updateStatus(myMessage.getMessageId(), BrokerMessageStatus.SEND_FAIL);
            }
        }
    }

    /**
     * $sendKernel 发送消息的核心方法，使用异步线程池进行发送消息
     *
     * @param myMessage 消息
     */
    private void sendKernel(MyMessage myMessage) {
        ThreadUtils.asyncSumbit(() -> {
            CorrelationDataExt correlationData = new CorrelationDataExt(
                    String.format("%s#%s", myMessage.getMessageId(), myMessage.getMessageType()),
                    myMessage.getTopic(), myMessage.getRoutingKey(), myMessage.getData()
            );
            RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getTemplate(myMessage);
            try {
                log.info("#发送MQ新消息#, 消息ID: {}", myMessage.getMessageId());
                rabbitTemplate.convertAndSend(myMessage.getTopic(), myMessage.getRoutingKey(), myMessage, messagePostProcessor -> {
                    if (myMessage.getDelayMills() != null) {
                        messagePostProcessor.getMessageProperties().setHeader("x-delay", myMessage.getDelayMills() * 1000);
                    }
                    messagePostProcessor.getMessageProperties().setContentType("application/json");
                    return messagePostProcessor;
                }, correlationData);
            } catch (Throwable e) {
                log.error("#发送MQ新消息失败#, 消息:{} 异常:{}", JSONObject.toJSON(myMessage), e.getMessage(), e);
                if (Objects.equals(myMessage.getMessageType(), MessageType.RELIANT)) {
                    messageStoreService.updateStatus(myMessage.getMessageId(), BrokerMessageStatus.SEND_FAIL);
                }
            }
        });

    }

}
